package com.hery.demo;

import com.hery.functions.SinkFunctions;
import com.hery.pojo.KafkaAlertLog;
import com.hery.pojo.Rule;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


public class KafkaSinkDemo {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> source = env.readTextFile("data/kafka.log");

        SingleOutputStreamOperator<KafkaAlertLog> filter = source.map(s -> new KafkaAlertLog(s, new Rule())).filter(KafkaAlertLog::isSentryAlertData);

        filter.map(KafkaAlertLog::getJson).addSink(SinkFunctions.getFlinkKafkaProducer());

        env.execute("kafka sink");

    }
}
